﻿using RabbitMQ.Client;
using RabbitMQPlugin;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace RabbitMQPublisher
{
    public class RabbitMQPublisher1
    {
        public static void PublishDefaulte() {
            using (var con = new RabbitMQHelper().GetConnection()) {
                using (var cl = con.CreateModel()) {
                    string message = "default message";
                    IBasicProperties prop = cl.CreateBasicProperties();
                    prop.ContentType = "text/plain";
                    prop.DeliveryMode = 2;//持久化消息
                    for (var i = 0; i < 100; i++)
                    {
                        Console.WriteLine($"message{i}");
                        cl.BasicPublish("", "fanout_queue_001", prop, Encoding.UTF8.GetBytes(message));
                    }
                }
            }
        }

        /// <summary>
        /// Dirct消息发布(持久化、手动消除)
        /// </summary>
        public static void PublisDirct() {
            using (var conn = new RabbitMQHelper().GetConnection()) {
                using (var channel = conn.CreateModel())
                {
                    string exchangeName = "direct_exchange_001";
                    string ququeName = "direct_queque_001-1";
                    string routingKey = "direct_routekey_001-1";

                    channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: true, autoDelete: false);
                    Dictionary<string, object> props = new Dictionary<string, object>();
                    //props.Add("x-message-ttl", 1000 * 600);//设置队列过期时间
                    channel.QueueDeclare(ququeName, durable: true, exclusive: false, autoDelete: false, arguments: props);
                    channel.QueueBind(ququeName, exchangeName, routingKey);

                    IBasicProperties prop = channel.CreateBasicProperties();
                    prop.ContentType = "text/plain";
                    prop.DeliveryMode = 2;//持久化消息
                    //prop.Expiration =$"{1000*6}";//过期时间

                    for (var i = 0; i < 1; i++)
                    {
                        string message = $"Dicrct_PublishMesssage_Time:{DateTime.Now.ToLongTimeString()}";
                        Console.WriteLine(message);

                        channel.ConfirmSelect();//开启确认模式
                        channel.BasicAcks += (sender, e) => {
                            //消息到达队列回调
                            Console.WriteLine("BasicAcks消息确认收到");
                        };

                        channel.BasicReturn += (sender, e) => {
                            //消息未到达队列
                            var rCode= e.ReplyCode;
                            var rText = e.ReplyText;
                            ReadOnlySpan<byte> span = e.Body.Span;
                            string message = Encoding.UTF8.GetString(span);
                        };


                        channel.BasicNacks += (sender, e) => { 
                            //拒绝签收回调
                        };
                        channel.BasicPublish(exchangeName, routingKey, prop, Encoding.UTF8.GetBytes(message));

                        if (channel.WaitForConfirms())
                        {
                            Console.WriteLine("消息到达");
                        }
                        else {
                            //消息未到达队列
                            Console.WriteLine("消息发送失败");
                            break;
                        }


                        //Thread.Sleep(1000);
                    }
                }
            }
        }

        public static void PublisFanout()
        {
            using (var con = new RabbitMQHelper().GetConnection())
            {
                using (var channel = con.CreateModel())
                {
                    string exchangeName = "fanout_exchange_001";

                    channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout, durable: true, autoDelete: false, null);

                    var arguments = new Dictionary<string, object>();
                    channel.QueueDeclare("fanout_queue_001", durable: true, autoDelete: false, exclusive: false, arguments: arguments);
                    channel.QueueDeclare("fanout_queue_002", durable: true, autoDelete: false, exclusive: false, arguments: arguments);
                    channel.QueueDeclare("fanout_queue_003", durable: true, autoDelete: false, exclusive: false, arguments: arguments);

                    channel.QueueBind("fanout_queue_001", exchangeName, "", null);
                    channel.QueueBind("fanout_queue_002", exchangeName, "", null);
                    channel.QueueBind("fanout_queue_003", exchangeName, "", null);

                    var props = channel.CreateBasicProperties();
                    props.DeliveryMode = 2;//消息持久化
                    for (var i = 0; i < 1; i++)
                    {
                        string message = $"FanOut_PublishMesssage_Time:{DateTime.Now.ToLongTimeString()}";
                        Console.WriteLine(message);

                        channel.BasicPublish(exchangeName, "", props, Encoding.UTF8.GetBytes(message));
                        //Thread.Sleep(200);
                    }
                }
            }
        }

        public static void PublisTopic() {
            using (var con = new RabbitMQHelper().GetConnection()) {
                using (var channel = con.CreateModel()) {
                    string exchangeName = "topic_exchange_001";
                    string queuename1 = "topic_exchange.add";
                    string queuename2 = "topic_exchange.update";
                    string queuename3 = "topic_exchange.*";
                    channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, durable: true, autoDelete: false, null);

                    channel.QueueDeclare(queuename1, durable: true, exclusive: false, autoDelete: false, null);
                    channel.QueueDeclare(queuename2, durable: true, exclusive: false, autoDelete: false, null);
                    channel.QueueDeclare(queuename3, durable: true, exclusive: false, autoDelete: false, null);

                    channel.QueueBind(queuename1, exchangeName, queuename1, null);
                    channel.QueueBind(queuename2, exchangeName, queuename2, null);
                    channel.QueueBind(queuename3, exchangeName, queuename3, null);

                    string message = $"Topic_xxx_Time:{DateTime.Now.ToLongTimeString()}";
                    Console.WriteLine(message);

                    var pp= channel.CreateBasicProperties();
                    pp.DeliveryMode = 2;
                    channel.BasicPublish(exchangeName, "topic_exchange.update", pp, Encoding.UTF8.GetBytes(message));
                }
            }
        }

        /// <summary>
        /// 测试死信队列
        /// </summary>
        public static void PublisDirectDLL()
        {
            string exchangename = "direct_exchagne_ddltext_001";
            string queuename = exchangename;
            string routingkey = exchangename;

            string exchangenameD = "direct_exchagne_ddl_003";
            string queuenameD = exchangenameD;
            string routingkeyD = exchangenameD;
            using (var con = new RabbitMQHelper().GetConnection())
            {
                using (var channel = con.CreateModel())
                {
                    //申明死信队列
                    channel.ExchangeDeclare(exchangenameD, ExchangeType.Direct, durable: true, autoDelete: false);
                    channel.QueueDeclare(queuenameD, durable: true, exclusive: false, autoDelete: false, null);
                    channel.QueueBind(queuenameD, exchangenameD, routingkeyD, null);

                    //申明队列
                    channel.ExchangeDeclare(exchangename, ExchangeType.Direct, durable: true, autoDelete: false);
                    var args = new Dictionary<string, object>();
                    args.Add("x-dead-letter-exchange", exchangenameD);
                    args.Add("x-dead-letter-routing-key", routingkeyD);
                    args.Add("x-message-ttl", 10000);
                    channel.QueueDeclare(queuename, durable: true, exclusive: false, autoDelete: false, args);
                    channel.QueueBind(queuename, exchangename, routingkey, null);

                    var pps = channel.CreateBasicProperties();
                    //pps.DeliveryMode = 2;
                    pps.Persistent = true;
                    //pps.Expiration = $"{1000 * 10}";
                    for (var i = 0; i < 100; i++)
                    {
                        Console.WriteLine($"发送测试死信队列{i}消息");
                        channel.BasicPublish(exchangename, routingkey, pps, Encoding.UTF8.GetBytes($"测试死信队列,Datetiem:{DateTime.Now.ToString()}"));
                    }

                }
            }
        }

        /// <summary>
        /// 测试死信队列
        /// </summary>
        public static void PublisFanoutDLLSSS()
        {
            string exchangename = "dirct.exchagne.test.for.ddl";
            string queuename = exchangename;
            string routingkey = exchangename;

            string exchangenameD = "dirct.exchagne.ddl";
            string queuenameD = exchangenameD;
            string routingkeyD = exchangenameD;
            using (var con = new RabbitMQHelper().GetConnection())
            {
                using (var channel = con.CreateModel())
                {
                    channel.ExchangeDeclare(exchangename, ExchangeType.Direct, durable: true, autoDelete: false);
                    var args = new Dictionary<string, object>();
                    args.Add("x-dead-letter-exchange", exchangenameD);
                    args.Add("x-dead-letter-routing-key", routingkeyD);
                    //args.Add("x-message-ttl", 10000);
                    channel.QueueDeclare(queuename, durable: true, exclusive: false, autoDelete: false, args);
                    channel.QueueBind(queuename, exchangename, routingkey, null);

                    channel.ExchangeDeclare(exchangenameD, ExchangeType.Direct, durable: true, autoDelete: false);
                    channel.QueueDeclare(queuenameD, durable: true, exclusive: false, autoDelete: false, null);
                    channel.QueueBind(queuenameD, exchangenameD, routingkeyD, null);

                    var pps = channel.CreateBasicProperties();
                    //pps.DeliveryMode = 2;
                    pps.Persistent = true;
                    pps.Expiration = $"{1000 * 10}";
                    for (var i = 0; i < 100; i++)
                    {
                        channel.BasicPublish(exchangename, routingkey, pps, Encoding.UTF8.GetBytes($"测试死信队列,Datetiem:{DateTime.Now.ToString()}"));
                    }
                }
            }
        }

        public static void 校验队列是否存在() {
            using (var conn = new RabbitMQHelper().GetConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    string ququeName = "direct_queque_001";

                    QueueDeclareOk declareOk = channel.QueueDeclarePassive(ququeName);
                }
            }
        }
    }
}
